package com.bite.messageQueue.mqServer;

import com.bite.messageQueue.common.Consumer;
import com.bite.messageQueue.common.MqException;
import com.bite.messageQueue.mqServer.core.*;
import com.bite.messageQueue.mqServer.datacenter.DiskDataCenter;
import com.bite.messageQueue.mqServer.datacenter.MemoryDataCenter;
import lombok.extern.slf4j.Slf4j;

import java.io.IOException;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * 虚拟主机
 * 每个虚拟主机都有自己的交换机，队列，绑定关系，消息 等数据
 * 提供接口给上层调用
 */
@Slf4j
public class VirtualHost {
    private String virtualHostname;
    private MemoryDataCenter memoryDataCenter = new MemoryDataCenter();
    private DiskDataCenter diskDataCenter = new DiskDataCenter();
    private Router router = new Router();
    private ConsumerManager consumerManager = new ConsumerManager(this);

    //操作交换机的锁对象
    private final Object exchangeLock = new Object();
    //操作队列的锁对象
    private final Object queueLock = new Object();

    public VirtualHost(String name) {
        this.virtualHostname = name;
        //从硬盘上初始化数据库文件
        diskDataCenter.init();

        //加载数据进内存
        try {
            memoryDataCenter.recovery(diskDataCenter);
        } catch (IOException | ClassNotFoundException | MqException e) {
            e.printStackTrace();
            log.error("[" + virtualHostname + "]虚拟主机恢复数据失败");
        }
    }

    /**
     * 交换机操作
     */
    public boolean exchangeDeclare(String exchangeName, ExchangeType exchangeType, boolean durable, boolean autoDelete,
                                   Map<String, Object> arguments) throws MqException {
        //构建交换机名称
        exchangeName = virtualHostname + exchangeName;
        try {
            synchronized (exchangeLock) {
                Exchange existsExhange = memoryDataCenter.getExchange(exchangeName);
                if (existsExhange != null) {
                    log.info("交换机" + existsExhange + "已存在");
                    throw new MqException("MqException:交换机" + exchangeName + "已存在");
                }
                //创建交换机对象
                Exchange exchange = new Exchange();
                exchange.setName(exchangeName);
                exchange.setExchangeType(exchangeType);
                exchange.setDurable(durable);
                exchange.setAutoDelete(autoDelete);
                exchange.setArguments(arguments);
                if (durable) {
                    //写入硬盘
                    diskDataCenter.insertExchange(exchange);
                }
                //写入内存
                memoryDataCenter.insertExchange(exchange);
                log.info("交换机" + exchange + "创建成功");
            }
            return true;
        } catch (Exception e) {
            log.error("交换机" + exchangeName + "创建失败");
            return false;
        }
    }

    public boolean exchangeDelete(String exchangeName) throws MqException {
        //构建交换机名称
        exchangeName = virtualHostname + exchangeName;
        try {
            synchronized (exchangeLock) {
                Exchange exchange = memoryDataCenter.getExchange(exchangeName);
                if (exchange == null) {
                    log.error("交换机" + exchangeName + "不存在");
                    throw new MqException("MqException:交换机" + exchangeName + "不存在");
                }
                if (exchange.isDurable()) {
                    diskDataCenter.deleteExchange(exchangeName);
                }
                memoryDataCenter.deleteExchange(exchangeName);
                log.info("交换机" + exchangeName + "删除成功");
            }
            return true;
        } catch (Exception e) {
            log.error("交换机" + exchangeName + "删除失败");
            return false;
        }
    }

    /**
     * 队列操作
     */
    public boolean queueDeclare(String queueName, boolean durable, boolean exclusive, boolean autoDelete,
                                Map<String, Object> arguments) throws MqException, IOException {
        //构建队列名称
        queueName = virtualHostname + queueName;
        try {
            synchronized (queueLock) {
                if (memoryDataCenter.getQueue(queueName) != null) {
                    log.info("队列" + queueName + "已存在");
                    throw new MqException("MqException:队列" + queueName + "已存在");
                }
                // 创建队列对象
                MSGQueue queue = new MSGQueue();
                queue.setName(queueName);
                queue.setDurable(durable);
                queue.setExclusive(exclusive);
                queue.setAutoDelete(autoDelete);
                queue.setArguments(arguments);
                if (durable) {
                    diskDataCenter.insertQueue(queue);
                }
                memoryDataCenter.insertQueue(queue);
                log.info("队列" + queue + "创建成功");
            }
            return true;
        } catch (Exception e) {
            log.error("队列" + queueName + "创建失败");
            return false;
        }
    }

    public boolean queueDelete(String queueName) throws MqException {
        //构建队列名称
        queueName = virtualHostname + queueName;
        try {
            synchronized (queueLock) {
                MSGQueue queue = memoryDataCenter.getQueue(queueName);
                if (queue == null) {
                    log.error("队列" + queueName + "不存在");
                    throw new MqException("MqException:队列" + queueName + "不存在");
                }
                if (queue.isDurable()) {
                    diskDataCenter.deleteQueue(queueName);
                }
                memoryDataCenter.deleteQueue(queueName);
                log.info("队列" + queueName + "删除成功");
            }
            return true;
        } catch (Exception e) {
            log.error("队列" + queueName + "删除失败");
            return false;
        }
    }

    /**
     * 绑定操作
     */
    public boolean queueBind(String queueName, String exchangeName, String bindingKey) throws MqException {
        //构建队列和交换机名称
        queueName = virtualHostname + queueName;
        exchangeName = virtualHostname + exchangeName;
        try {
            synchronized (exchangeLock) {
                synchronized (queueLock) {
                    //检查绑定是否已存在
                    Binding exitsBinding = memoryDataCenter.getBinding(queueName, exchangeName);
                    if (exitsBinding != null) {
                        log.info("绑定" + exitsBinding + "已存在");
                        throw new MqException("MqException:绑定" + exitsBinding + "已存在");
                    }
                    //验证BindingKey是否合法
                    if (!router.checkBindingKey(bindingKey)) {
                        log.error("绑定key不合法");
                        throw new MqException("MqException:绑定key不合法");
                    }
                    //创建绑定对象
                    Binding binding = new Binding();
                    binding.setQueueName(queueName);
                    binding.setExchangeName(exchangeName);
                    binding.setBindingKey(bindingKey);
                    //获取对应的交换机和队列
                    MSGQueue queue = memoryDataCenter.getQueue(queueName);
                    Exchange exchange = memoryDataCenter.getExchange(exchangeName);
                    if (queue == null) {
                        log.error("队列" + queueName + "不存在");
                        throw new MqException("MqException:队列" + queueName + "不存在");
                    }
                    if (exchange == null) {
                        log.error("交换机" + exchangeName + "不存在");
                        throw new MqException("MqException:交换机" + exchangeName + "不存在");
                    }
                    //写入硬盘
                    if (queue.isDurable() && exchange.isDurable()) {
                        diskDataCenter.insertBinding(binding);
                    }
                    //写入内存
                    memoryDataCenter.insertBinding(binding);
                    log.info("绑定" + exchangeName + " : " + queueName + "成功");
                }
            }
            return true;
        } catch (Exception e) {
            log.error("绑定" + exchangeName + " : " + queueName + "失败");
            return false;
        }
    }

    public boolean queueUnbind(String queueName, String exchangeName) throws MqException {
        //构建队列和交换机名称
        queueName = virtualHostname + queueName;
        exchangeName = virtualHostname + exchangeName;
        try {
            synchronized (exchangeLock) {
                synchronized (queueLock) {
                    //检查绑定是否已存在
                    Binding exitsBinding = memoryDataCenter.getBinding(exchangeName, queueName);
                    if (exitsBinding == null) {
                        log.info("绑定" + exitsBinding + "不存在");
                        throw new MqException("MqException:绑定" + exitsBinding + "不存在");
                    }
                    //删除硬盘上的绑定
                    diskDataCenter.deleteBinding(exitsBinding);
                    //删除内存中的绑定
                    memoryDataCenter.deleteBinding(exitsBinding);
                    log.info("绑定" + exchangeName + " : " + queueName + "删除成功");
                }
            }
            return true;
        } catch (Exception e) {
            log.error("绑定" + exchangeName + " : " + queueName + "删除失败");
            return false;
        }
    }

    /**
     * 发送消息到指定交换机/队列
     */
    public boolean basicPublish(String exchangeName, String routingKey,
                               BasicProperties basicProperties, byte[] body) throws MqException {
        try {
            //检查routingKey是否合法
            if (!router.checkRoutingKey(routingKey)) {
                log.error("routingKey不合法");
                throw new MqException("MqException:routingKey不合法");
            }
            //构建交换机名称
            exchangeName = virtualHostname + exchangeName;
            //查找交换机对象
            Exchange exchange = memoryDataCenter.getExchange(exchangeName);
            if (exchange == null) {
                log.error("交换机" + exchangeName + "不存在");
                throw new MqException("MqException:交换机" + exchangeName + "不存在");
            }
            //判定交换机类型
            if (exchange.getExchangeType() == ExchangeType.DIRECT) {
                // 按照直接交换机的方式来转发消息
                // 以 routingKey 作为队列的名字, 直接把消息写入指定的队列中.
                String queueName = virtualHostname + routingKey;
                // 5. 构造消息对象
                Message message = Message.createMessageWithID(routingKey, basicProperties, body);
                // 6. 查找该队列名对应的对象
                MSGQueue queue = memoryDataCenter.getQueue(queueName);
                if (queue == null) {
                    throw new MqException("[VirtualHost] 队列不存在! queueName=" + queueName);
                }
                // 7. 队列存在, 直接给队列中写入消息
                sendMessage(queue, message);
            } else {
                // 按照 fanout 和 topic 的方式来转发.
                // 5. 找到该交换机关联的所有绑定, 并遍历这些绑定对象
                ConcurrentHashMap<String, Binding> bindingsMap = memoryDataCenter.getBindings(exchangeName);
                for (Map.Entry<String, Binding> entry : bindingsMap.entrySet()) {
                    // 1) 获取到绑定对象, 判定对应的队列是否存在
                    Binding binding = entry.getValue();
                    MSGQueue queue = memoryDataCenter.getQueue(binding.getQueueName());
                    if (queue == null) {
                        // 此处不抛出异常. 可能此处有多个这样的队列.
                        System.out.println("[VirtualHost] basicPublish 发送消息时, 发现队列不存在! queueName=" + binding.getQueueName());
                        continue;
                    }
                    // 2) 构造消息对象
                    Message message = Message.createMessageWithID(routingKey, basicProperties, body);
                    // 3) 判定这个消息是否能转发给该队列.
                    //    如果是 fanout, 所有绑定的队列都要转发的.
                    //    如果是 topic, 还需要判定下, bindingKey 和 routingKey 是不是匹配.
                    if (!router.route(exchange.getExchangeType(), binding, message)) {
                        continue;
                    }
                    // 4) 真正转发消息给队列
                    sendMessage(queue, message);
                }
            }
            return true;
        } catch (Exception e) {
            log.error("发送消息到" + exchangeName + "失败");
            return false;
        }
    }

    //将消息写入硬盘和内存
    private void sendMessage(MSGQueue queue, Message message) throws IOException, MqException, InterruptedException {
        int deliverMode = message.getDeliverMode();
        if(deliverMode == 2){
            //持久化到硬盘
            diskDataCenter.sendMessage(queue, message);
        }
        //写入内存
        memoryDataCenter.sendMessageToQueue(queue, message);

        // 此处还需要补充一个逻辑, 通知消费者可以消费消息了.
        consumerManager.notifyConsume(queue.getName());
    }

    /**
     * 消费者订阅
     * @param consumerTag 消费者标识
     * @param queueName 要订阅的队列名称
     * @param autoACK 自动应答
     * @param consumer 消费者
     */
    public boolean basicConsume(String consumerTag, String queueName,
                                 boolean autoACK, Consumer consumer){
        //构造一个ConsumerEnv对象，找到其对应的队列，再把它添加到队列中
        queueName = virtualHostname + queueName;
        try{
            consumerManager.addConsumer(consumerTag, queueName, autoACK, consumer);
            log.info("消费者" + consumerTag + "订阅队列" + queueName + "成功");
            return true;
        } catch (Exception e) {
            log.error("[VirtualHost] basicConsume 失败! queueName=" + queueName);
            e.printStackTrace();
            return false;
        }
    }

    //自动应答
    public boolean basicACK(String queueName, String messageId){
        queueName = virtualHostname + queueName;
        try{
            Message message = memoryDataCenter.getMessage(messageId);
            if(message == null){
                log.error("[VirtualHost] basicACK 失败! queueName=" + queueName + " messageId=" + messageId);
                throw new MqException("MqException:消息不存在");
            }
            MSGQueue queue = memoryDataCenter.getQueue(queueName);
            if(queue == null){
                log.error("[VirtualHost] basicACK 失败! queueName=" + queueName + " messageId=" + messageId);
                throw new MqException("MqException:队列不存在");
            }
            //删硬盘
            if(message.getDeliverMode() == 2){
                diskDataCenter.deleteMessage(queue, message);
            }
            //删待确认队列
            memoryDataCenter.removeMessageWaitACK(queueName, messageId);
            //删内存
            memoryDataCenter.removeMessage(message.getMessageID());
            log.info("消息被手动应答的方式消费");
            return true;
        } catch (Exception e){
            log.error("[VirtualHost] basicACK 失败! queueName=" + queueName + " messageId=" + messageId);
            e.printStackTrace();
            return false;
        }
    }

    /**
     * 获取
     *
     * @return virtualHostname
     */
    public String getVirtualHostname() {
        return virtualHostname;
    }

    /**
     * 设置
     *
     * @param virtualHostname
     */
    public void setVirtualHostname(String virtualHostname) {
        this.virtualHostname = virtualHostname;
    }

    /**
     * 获取
     *
     * @return memoryDataCenter
     */
    public MemoryDataCenter getMemoryDataCenter() {
        return memoryDataCenter;
    }

    /**
     * 设置
     *
     * @param memoryDataCenter
     */
    public void setMemoryDataCenter(MemoryDataCenter memoryDataCenter) {
        this.memoryDataCenter = memoryDataCenter;
    }

    /**
     * 获取
     *
     * @return diskDataCenter
     */
    public DiskDataCenter getDiskDataCenter() {
        return diskDataCenter;
    }

    /**
     * 设置
     *
     * @param diskDataCenter
     */
    public void setDiskDataCenter(DiskDataCenter diskDataCenter) {
        this.diskDataCenter = diskDataCenter;
    }
}
